package producer

import (
	"bp/conf"
	"bp/db"
	sqlc "bp/db/sqlc"
	"bp/internal/nsqserver/shared"
	"context"
	"fmt"

	"github.com/nsqio/go-nsq"
	"go.uber.org/zap"
)

const (
	topicOfOrderOffApportion = "order_off_apportion"
)

// orderOffApportion topicOfOrderOffApportion's body
type OffApportionNotification struct {
	TransNo     string                        `json:"out_trans_no"`
	OrderNo     string                        `json:"out_order_no"`
	AfterSaleNo string                        `json:"out_after_sale_no"`
	Details     []*ApportionmentOfOrderDetail `json:"details"`
	Timestamp   int64                         `json:"timestamp"`
}

type ApportionmentOfOrderDetail struct {
	OutTransNo           string `json:"out_trans_no"`
	OutOrderNo           string `json:"out_order_no"`
	OutAfterSaleNo       string `json:"out_after_sale_no"`
	OutOrderDetailID     int64  `json:"out_order_detail_id"`
	OutAfterSaleDetailID int64  `json:"out_after_sale_detail_id"`
	Off                  int32  `json:"off"`
	Cnt                  int32  `json:"cnt"`
}

func makeProducerTopic(topic string) string {
	return fmt.Sprintf("%s-%s", shared.AppID, topic)
}

// NsqProducerProxy publish nsq message producer
type NsqProducerProxy struct {
	slog     *zap.SugaredLogger
	pgdao    *db.PgDao
	producer *nsq.Producer
}

// NewNsqProducerProxy initialization
func NewNsqProducerProxy(
	slog *zap.SugaredLogger,
	pgdao *db.PgDao,
	nsqConfPtr *conf.NsqSection,
) (proxy *NsqProducerProxy, err error) {
	proxy = &NsqProducerProxy{
		slog:  slog.With("module", "NsqProducerProxy"),
		pgdao: pgdao,
	}
	config := nsq.NewConfig()

	producer, err := nsq.NewProducer(nsqConfPtr.NsqdAddress, config)
	if err != nil {
		return
	}

	nsqLogger := shared.NewNoopNSQLogger(slog)

	proxy.producer = producer
	producer.SetLogger(
		nsqLogger,
		nsq.LogLevelError,
	)

	// Gracefully stop the consumer.
	// consumer.Stop()

	return
}

// func (s *NsqProducerProxy) UpsertNsqPub(topic, sn string, messageBytes []byte) (nsqPubPtr *sqlc.NsqPub, err error) {
// 	ctx := context.Background()
// 	q := s.pgdao.Q
// 	nsqPub, err := q.UpsertNsqPub(ctx, sqlc.UpsertNsqPubParams{
// 		Topic:   topic,
// 		Sn:      sn,
// 		Message: messageBytes,
// 	})
// 	if err != nil {
// 		return
// 	}
// 	nsqPubPtr = &nsqPub
// 	return
// }

func (s *NsqProducerProxy) HandleNsqPubFail(nsqPubID int64, errPtr *error) {
	if errPtr != nil && (*errPtr) != nil {
		ctx := context.Background()
		q := s.pgdao.Q
		err2 := q.UpdateNsqPubError(ctx, sqlc.UpdateNsqPubErrorParams{
			ID:     nsqPubID,
			ErrMsg: sqlc.NSFromString((*errPtr).Error()),
		})
		if err2 != nil {
			s.slog.Errorf("HandleNsqPubFail err: %s", err2.Error)
		}
	}
}

func (s *NsqProducerProxy) GetPublishOffApportionmentsTopic() string {
	return makeProducerTopic(topicOfOrderOffApportion)
}

func (s *NsqProducerProxy) PublishNotification(ptr *sqlc.NsqPub) (err error) {
	defer s.HandleNsqPubFail(ptr.ID, &err)
	err = s.producer.Publish(ptr.Topic, ptr.Message)
	if err != nil {
		return
	}
	return
}
